package rpc.core;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rpc.core.registry.RegistryManager;
import rpc.netty.tcp.NettyClient;
import rpc.netty.tcp.entity.RpcMessage;
import rpc.trace.Trace;
import rpc.trace.TraceContext;
import rpc.utils.ExceptionUtils;
import rpc.utils.ReflectionUtils;

/*********
 *
 *
 *@author 197
 *@date 2020/7/10 14:42
 *
 **/
public class RefererProxy<T> implements InvocationHandler {

  private static final Logger logger = LoggerFactory.getLogger(RefererProxy.class);
  volatile int currIndex = 0;
  private Class<?> interfaceClass;
  private int serviceId = 0;

  @SuppressWarnings("unchecked")
  public T bind(Class<?> cls) {
    this.interfaceClass = cls;
    this.serviceId = ReflectionUtils.getFliedIntValue(interfaceClass, "serviceId");
    return (T) Proxy.newProxyInstance(cls.getClassLoader(), new Class[]{interfaceClass}, this);
  }

  @SuppressWarnings("SingleStatementInBlock")
  @Override
  public Object invoke(Object proxy, Method method, Object[] args) {
    if (logger.isDebugEnabled()) {
      logger.debug("interfaceClass:{},proxy:{}，method:{},args[0]:{}", interfaceClass.getSimpleName(), proxy.getClass().getName(), method.getName(), args[0]);
    }
    if (Objects.isNull(args)) {
      return new FlowRes().fail(-823);
    }

    TraceContext traceContext = Trace.get();
    String sequence = traceContext.getTraceId();
    FlowReq req = (FlowReq) args[0];
    CompletableFuture<FlowRes> f = CompletableFuture.supplyAsync(() -> {

      List<NettyClient> nettyClientList = RegistryManager.nettyClientListMap.get(interfaceClass);
      if (CollectionUtils.isEmpty(nettyClientList)) {
        return new FlowRes().fail(-812);
      }
      nettyClientList.get(currIndex).writeAndFlush(serviceId,
          ReflectionUtils.getFliedIntValue(interfaceClass, method.getName() + "_MsgId"),
          sequence, req);
      currIndex++;
      if (nettyClientList.size() >= currIndex) {
        currIndex = 0;
      }
      BlockingQueue<RpcMessage> rpcMessages = NettyClient.messageQueueMap.get(sequence);
      RpcMessage message = null;

      try {
        message = rpcMessages.take();
      } catch (Exception e) {
        logger.error("exception:{} stackTrace:{}", e.getMessage(), ExceptionUtils.getTrace(e));
      }

      return new FlowRes(message.getData());
    });
    NettyClient.messageQueueMap.remove(sequence);

    try {
      return f.get(req.getTimeOut(), TimeUnit.MILLISECONDS);
    } catch (InterruptedException | ExecutionException e) {
      logger.error("exception:{} stackTrace:{}", e.getMessage(), ExceptionUtils.getTrace(e));
      return new FlowRes().fail(-824);
    } catch (TimeoutException e) {
      logger.error("exception:{} stackTrace:{}", e.getMessage(), ExceptionUtils.getTrace(e));
      return new FlowRes().fail(-818);
    } catch (Exception e) {
      logger.error("exception:{} stackTrace:{}", e.getMessage(), ExceptionUtils.getTrace(e));
      return new FlowRes().fail(-825);
    }
  }
}